package com.future.service.Impl;

import cn.hutool.json.JSONUtil;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.future.allUtils.constants.RedisCacheConstants;
import com.future.allUtils.utils.CacheClient;
import com.future.allUtils.utils.SecurityUtils;
import com.future.domain.ChatMessage;
import com.future.domain.DTO.MessageDTO;
import com.future.domain.DTO.MessagePageDTO;
import com.future.domain.Result;
import com.future.mapper.ChatMessageMapper;
import com.future.mapper.ChatRoomMemberMapper;
import com.future.service.ChatMemberService;
import com.future.service.ChatMessageService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.security.core.parameters.P;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@Service
@Slf4j
public class ChatMessageImpl implements ChatMessageService {
    @Autowired
    private CacheClient cacheClient;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private ChatMessageMapper chatMessageMapper;
    @Autowired
    private ChatRoomMemberMapper roomMemberMapper;
    @Autowired
    private ChatMemberService memberService;
    @Autowired
    private ObjectMapper objectMapper;
    @Override
    public void saveChatMessage(MessageDTO chatMessage) {
        try {
            if (chatMessage == null || chatMessage.getChatRoomId() == null || chatMessage.getSenderId() == null) {
                return;
            }
            //消息存入数据库
            ChatMessage chatMessage1 = new ChatMessage();
            BeanUtils.copyProperties(chatMessage, chatMessage1);
            chatMessage1.setSenderId(Long.valueOf(chatMessage.getSenderId()));
            chatMessage1.setChatRoomId(Long.valueOf(chatMessage.getChatRoomId()));
            chatMessageMapper.insert(chatMessage1);
            Long messageId=chatMessage1.getId();
            chatMessage.setId(messageId.toString());
            Long roomId = chatMessage1.getChatRoomId();
            //获取聊天室成员
            //从redis中
            String memberKey = RedisCacheConstants.CHAT_ROOM_MEMBERS_PREFIX + roomId;
            Set<String> members = stringRedisTemplate.opsForSet().members(memberKey);
            if (members == null || members.isEmpty()) {
                members = Collections.emptySet();
            }
            List<Long> userIds = members.stream()
                    .filter(Objects::nonNull)
                    .filter(userId -> !userId.equals(chatMessage.getSenderId()))
                    .map(Long::valueOf)
                    .collect(Collectors.toList());
            //redis未命中，从数据库中
            if (userIds.isEmpty()) {
                userIds = roomMemberMapper.selectUserIdByRoomId(roomId).stream()
                        .filter(userId -> !userId.equals(Long.valueOf(chatMessage.getSenderId()))) // 排除发送者
                        .collect(Collectors.toList());
            }
            //批量添加用户的未读消息数据
            chatMessageMapper.batchInsert(messageId, roomId, userIds);
            //把消息存入redis
            String messagekey = RedisCacheConstants.CHAT_MESSAGE_PREFIX + roomId;
            String message = null;
            try {
                message = objectMapper.writeValueAsString(chatMessage);
            } catch (JsonProcessingException e) {
                log.error("序列化消息失败，消息ID:{}", chatMessage1.getId(), e);
            }
            stringRedisTemplate.opsForZSet().add(messagekey, message, chatMessage.getSendTime().getTime());
            stringRedisTemplate.expire(messagekey, 30, TimeUnit.MINUTES);
            // 4. 处理离线消息（判断接收者是否在线）
            handleOfflineMessage(chatMessage);
        }catch (Exception e)
        {
            log.error(""+e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public Result getMessages(String roomId1, String lastMessageId1, Integer pageSize) {
        try {
            if(roomId1==null||roomId1.trim().isEmpty())
            {
                return Result.fail("房间号为空");
            }
            Long roomId = Long.valueOf(roomId1);
            Long userId = SecurityUtils.getCurrentUserId();
            if (!memberService.checkUserInRoom(userId, roomId)) {
                return Result.fail("无权访问该聊天室消息");
            }
            if (pageSize == null || pageSize < 0) {
                pageSize = 20;
            }
            //最大不超过一百
            pageSize = Math.min(pageSize, 100);
            //获取聊天信息
            String messageKey = RedisCacheConstants.CHAT_MESSAGE_PREFIX + roomId;
            ZSetOperations<String, String> zSetOps = stringRedisTemplate.opsForZSet();
            // 获取ZSet总长度（用于计算索引）
            Long totalSize = zSetOps.size(messageKey);
            List<MessageDTO> messageDTOS = new ArrayList<>();
            Long end=0L;
            Long start=0L;
            String currentLastId = null;
            boolean hasMore = false;
            if (totalSize != null && totalSize > 0) {

                if (lastMessageId1 == null||lastMessageId1.trim().isEmpty()) {
                    end =Math.min(pageSize-1,totalSize-1);
                    start = 0L;
                } else {
                    Long lastMessageId=Long.valueOf(lastMessageId1);
                    MessageDTO messageDTO = chatMessageMapper.MessageDTO(lastMessageId);
                    String s = objectMapper.writeValueAsString(messageDTO);
                    //先定位到上一次最后一条消息的索引
                    Long rank = zSetOps.rank(messageKey, s);
                    if (rank != null) {
                        long reverseLastRank=totalSize-rank-1;
                        start=reverseLastRank+1;
                        end=start+pageSize-1;
                        end = Math.min(end, totalSize - 1);
                    }
                }
                    Set<ZSetOperations.TypedTuple<String>> typedTuples = zSetOps.reverseRangeWithScores(messageKey, start, end);
                    for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
                        String value = typedTuple.getValue();
                        MessageDTO messageDTO1= objectMapper.readValue(value, MessageDTO.class);
                        if (messageDTO1!= null) {
                            messageDTOS.add(messageDTO1);
                        }
                    }
                // 判断是否有更多消息
                hasMore = end < totalSize - 1;
                // 记录下一页的lastMessageId（当前页最后一条消息的ID，即反转前的第一条）
                if (!messageDTOS.isEmpty()) {
                    currentLastId =messageDTOS.get(messageDTOS.size() - 1).getId(); // 取列表最后一个元素的ID
                }

                }
            //redis未命中，从数据库里面查找
            if (messageDTOS.isEmpty()) {
                Long lastMessageId=(lastMessageId1==null||lastMessageId1.trim().isEmpty())?null:Long.valueOf(lastMessageId1);
                messageDTOS = chatMessageMapper.queryMessagesByCursor(roomId, lastMessageId, pageSize);
                //空值处理
                if (messageDTOS == null || messageDTOS.isEmpty()) {
                    return Result.ok("未查到聊天室消息");
                }
                //存入redis
                // 3. 遍历 List
                Set<ZSetOperations.TypedTuple<String>> tuples = messageDTOS.stream()
                        .map(message -> {
                            double score = message.getSendTime().getTime();
                            String value = null;
                            try {
                                value = objectMapper.writeValueAsString(message);
                            } catch (JsonProcessingException e) {
                                throw new RuntimeException(e);
                            }
                            return ZSetOperations.TypedTuple.of(value, score);
                        })
                        .filter(tuple -> tuple != null)
                        .collect(Collectors.toSet());
                ;
                if (!tuples.isEmpty()) {
                    zSetOps.add(messageKey, tuples);
                }
                stringRedisTemplate.expire(messageKey, 7, TimeUnit.DAYS);
                //计算当前页的最后一条消息id
               currentLastId =messageDTOS.get(messageDTOS.size() - 1).getId();
                //判断是否还有更多数据
                Integer remaining = chatMessageMapper.countMessagesAfterCursor(roomId, Long.valueOf(currentLastId));
               hasMore = remaining > 0;
            }
            Long currentUserId = SecurityUtils.getCurrentUserId();
            //封装数据
            for(MessageDTO messageDTO:messageDTOS)
            {
                Long senderId = Long.valueOf(messageDTO.getSenderId());
                boolean ismine=currentUserId.equals(senderId);
                messageDTO.setIsMine(ismine);
            }
            MessagePageDTO emptyPage = new MessagePageDTO();
            emptyPage.setMessages(messageDTOS);
            emptyPage.setLastMessageId(currentLastId);
            emptyPage.setHasMore(hasMore);
            chatMessageMapper.updateUnread(userId, roomId);
            String unreadKey = RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX + userId + ":" + roomId;
            stringRedisTemplate.delete(unreadKey);
            String offlineListKey = RedisCacheConstants.CHAT_OFFLINE_MESSAGE_LIST_PREFIX + userId + ":" + roomId;
            stringRedisTemplate.delete(offlineListKey);
            return Result.ok(emptyPage);
        }catch (Exception e) {
            log.error("获取历史消息失败", e);
            return Result.fail("获取历史消息失败：");
        }
    }


    @Override
    public Result getUnrendCountByRoom(String roomId1) {
        if(roomId1==null||roomId1.trim().isEmpty())
        {
            return Result.fail("房间号为空");
        }
        Long roomId=Long.valueOf(roomId1);
        Long userId=SecurityUtils.getCurrentUserId();
        try {
            if (!memberService.checkUserInRoom(userId, roomId)) {
                return Result.fail("无权访问该聊天室");
            }
            String unreadKey = RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX + userId + ":" + roomId;
            String countStr = stringRedisTemplate.opsForValue().get(unreadKey);
            // 若 Redis 缓存未命中，从数据库查询并回写缓存
            if (countStr == null||countStr.isEmpty()) {
                // 1. 从数据库查询未读消息数
                int dbCount = chatMessageMapper.countUnreadOne(userId, roomId);
                // 2. 回写 Redis 缓存
                stringRedisTemplate.opsForValue().set(unreadKey, String.valueOf(dbCount), 1, TimeUnit.HOURS);
                return Result.ok(dbCount);
            }
            // 缓存命中，直接返回
            return Result.ok(Integer.parseInt(countStr));
        } catch (Exception e) {
            log.error("获取未读消息数失败", e);
            return Result.fail("获取未读消息数失败：");
        }
    }

    @Override
    public Result getAllUnreadCounts() {
            Long userId = SecurityUtils.getCurrentUserId();
            if (userId == null) {
                return Result.fail("未获取到当前用户ID");
            }
            try {
                //批量获取redis缓存
                Map<Long, Integer> unreadMap = new HashMap<>();
                List<Long> roomIds = chatMessageMapper.roomList(userId);
                List<String> keys = roomIds.stream()
                        .map(roomId -> RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX + userId + ":" + roomId)
                        .collect(Collectors.toList());
                //批量查询未读消息数
                List<String> countStrs = stringRedisTemplate.opsForValue().multiGet(keys);
                //分离缓存命中和未命中的roomId
                List<Long> missRoomIds=new ArrayList<>();
                for(int i=0;i<roomIds.size();i++)
                {
                    String countStr = countStrs.get(i);
                    if(countStr==null||countStr.isEmpty())
                    {
                        missRoomIds.add(roomIds.get(i));
                    }else {
                        unreadMap.put(roomIds.get(i),Integer.parseInt(countStr));
                    }
                }
                if(!missRoomIds.isEmpty())
                {
                    //从数据库中查找
                    Map<Long, Integer> dbCounts = chatMessageMapper.countUnreadBatch(userId, missRoomIds);
                    //填充结果并回写reids
                    Map<String,String> redisMap=new HashMap<>();
                    for(Map.Entry<Long,Integer>entry:dbCounts.entrySet())
                    {
                        Long roomId= entry.getKey();
                        Integer count = entry.getValue();
                        unreadMap.put(roomId,count);
                        redisMap.put(RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX+userId+":"+roomId,String.valueOf(count));
                    }
                    //批量写入redis
                    if(!redisMap.isEmpty())
                    {
                        stringRedisTemplate.opsForValue().multiSet(redisMap);
                        redisMap.keySet().forEach(key->stringRedisTemplate.expire(key,1,TimeUnit.HOURS));
                    }
                }
                return Result.ok(unreadMap);
            } catch (Exception e) {
                log.error("获取所有未读消息数失败", e);
                return Result.fail("获取所有未读消息数失败：");
            }
        }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Result clearUnReadCount(String roomId1) {
        if(roomId1==null||roomId1.trim().isEmpty())
        {
            return Result.fail("房间号为空");
        }
        Long roomId=Long.valueOf(roomId1);
        Long userId=SecurityUtils.getCurrentUserId();
        try {
            if (!memberService.checkUserInRoom(userId, roomId)) {
                return Result.fail("无权操作该聊天室");
            }
            // 1. 清除Redis未读计数
            String unreadKey = RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX + userId + ":" + roomId;
            stringRedisTemplate.opsForValue().set(unreadKey, "0", 1, TimeUnit.HOURS);
            // 2. 标记该房间所有未读消息为已读
            chatMessageMapper.updateUnread(userId,roomId);
            return Result.ok("未读消息已清零");
        } catch (Exception e) {
            log.error("清零未读消息失败", e);
            return Result.fail("清零未读消息失败：");
        }
    }

    @Override
    public Result deleteChatMessage(String chatmessageId) {
        if(chatmessageId==null)
        {
            return Result.fail("消息id不能为空");
        }
        Long messageId = Long.valueOf(chatmessageId);
        ChatMessage chatMessage = chatMessageMapper.selectById(messageId);
        if(chatMessage==null)
        {
            return Result.fail("消息不存在");
        }
        Long roomId = chatMessageMapper.selectRoomIdBymessageId(messageId);
        if(roomId==null)
        {
            return Result.fail("聊天室不存在");
        }
        MessageDTO messageDTO = chatMessageMapper.MessageDTO(messageId);
        if(messageDTO==null)
        {

            return Result.ok("消息已删除");
        }
        String strjson=null;
        try {
            strjson = objectMapper.writeValueAsString(messageDTO);
            log.info(strjson);
        } catch (JsonProcessingException e) {
            Result.fail("消息序列化失败");
            throw new RuntimeException(e);
        }
        String messagekey = RedisCacheConstants.CHAT_MESSAGE_PREFIX + roomId;
        stringRedisTemplate.opsForZSet().remove(messagekey,strjson);
        //从数据库中删除
        chatMessageMapper.deleteMessage(messageId);
        return Result.ok("消息删除成功");
    }

    /**
     * 处理离线消息，接收者离线时存储到redis
     */
    public void handleOfflineMessage(MessageDTO message) {
        try {
        Long roomId = Long.valueOf(message.getChatRoomId());
        Long senderId = Long.valueOf(message.getSenderId());
        // 获取聊天室成员（除发送者自己）
        String memberKey = RedisCacheConstants.CHAT_ROOM_MEMBERS_PREFIX + roomId;
        Set<String> members = stringRedisTemplate.opsForSet().members(memberKey);
        if(members==null)
        {
            members=Collections.emptySet();
        }
        List<Long> userIds= members.stream()
                .filter(Objects::nonNull)  // 过滤null
                .map(Long::valueOf)
                .collect(Collectors.toList());
        //如果redis未命中，查询数据库
        if(userIds.isEmpty())
        {
            userIds= roomMemberMapper.selectUserIdByRoomId(roomId);
            if (userIds == null) {
                return;
            }
            //存入redis
            String[] array = userIds.stream()
                    .map(String::valueOf)
                    .toArray(String[]::new);
            stringRedisTemplate.opsForSet().add(memberKey,array);
            stringRedisTemplate.expire(memberKey, RedisCacheConstants.CACHE_NORMAL_TTL, TimeUnit.SECONDS);
        }
        log.info("当前聊天室 {} ", roomId);
        log.info("发送者 userId = {}", senderId);
        for (Long receiverId : userIds) {
            // 1. 存储离线消息（仅对离线用户，在线用户走实时推送）
            //排除发送者自己
            if(senderId.equals(receiverId))
            {
                continue;
            }
            String onlineKey = RedisCacheConstants.USER_ONLINE_PREFIX + receiverId;
            String onlineValue = stringRedisTemplate.opsForValue().get(onlineKey);
            if(onlineValue==null)
            {
                Integer c=roomMemberMapper.selectisOnlinebyuserId(receiverId);
                onlineValue = (c == null) ? "0" : String.valueOf(c);
            }
            boolean isOnline = "1".equals(onlineValue); // 只有值为"1"时才视为在线
            if (!isOnline) { // 离线（值为0）
                // 离线用户：存储消息到离线列表
                String offlineListKey = RedisCacheConstants.CHAT_OFFLINE_MESSAGE_LIST_PREFIX + receiverId+":"+roomId;
                String s = objectMapper.writeValueAsString(message);
                stringRedisTemplate.opsForList().leftPush(offlineListKey, s);
                stringRedisTemplate.expire(offlineListKey, 7, TimeUnit.DAYS);
                String offRoomListKey=RedisCacheConstants.CHAT_OFFLINE_MESSAGE_LIST_PREFIX+receiverId;
                stringRedisTemplate.opsForHash().put(offRoomListKey,String.valueOf(roomId),String.valueOf(System.currentTimeMillis()));
                stringRedisTemplate.expire(offRoomListKey,7,TimeUnit.DAYS);
            }

            // 2. 累加未读计数（关键：无论在线/离线，只要未读就计数）
            String unreadKey = RedisCacheConstants.CHAT_UNREAD_COUNT_PREFIX + receiverId + ":" + roomId;
            Long newCount = stringRedisTemplate.opsForValue().increment(unreadKey);
            stringRedisTemplate.expire(unreadKey, 7, TimeUnit.DAYS);
            log.info("用户[{}]在房间[{}]的未读计数+1，当前值：{}（消息ID：{}）",
                    receiverId, roomId, newCount, message.getId());
        } }catch (Exception e)
    {
        log.error(""+e);
    }
    }
}
